package cn._51doit.flink.day09;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

/**
 * 广播状态的使用场景：
 * 1.少量数据可以使用广播状态（广播状态是存储在内存中的，如果要广播的数据量非常大，不能使用广播状态）
 * 2.广播状态的数据可以动态改变
 *
 * //如果要广播的数据非常大，或者根本无法获取全部要广播的数据?
 * 为了高效的关联维度数据，可以使用异步IO
 *
 */
public class BroadcastStateDemo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //维度流
        //c100,电脑,INSERT
        //c101,手机,INSERT
        //c102,录音机,INSERT
        //c102,音响,UPDATE
        //c102,音响,DELETE
        DataStreamSource<String> lines1 = env.socketTextStream("localhost", 8888);

        //事实流(订单id,金额,分类id)
        //o001,3000,c100
        //o002,300,c102
        //o003,300,c102
        //o004,300,c102
        //o005,300,c102
        DataStreamSource<String> lines2 = env.socketTextStream("localhost", 9999);

        //期望得到的数据
        //o001,3000,c100,电脑

        SingleOutputStreamOperator<Tuple3<String, String, String>> tpStream1 = lines1.map(new MapFunction<String, Tuple3<String, String, String>>() {
            @Override
            public Tuple3<String, String, String> map(String line) throws Exception {
                String[] fields = line.split(",");
                return Tuple3.of(fields[0], fields[1], fields[2]);
            }
        });

        //在广播数据之前，需要定义，广播状态（就是要广播的数据，在下游以何种方式进行存储）
        //广播状态是一种特殊的OperatorState，使用之前，先定义一个状态描述器
        MapStateDescriptor<String, String> stateDescriptor = new MapStateDescriptor<>("category-state", String.class, String.class);
        //将数据广播到下游，并且将数据以Map形式保存在状态中
        BroadcastStream<Tuple3<String, String, String>> broadcastStream = tpStream1.broadcast(stateDescriptor);

        SingleOutputStreamOperator<Tuple3<String, Integer, String>> tpStream2 = lines2.map(new MapFunction<String, Tuple3<String, Integer, String>>() {
            @Override
            public Tuple3<String, Integer, String> map(String line) throws Exception {
                String[] fields = line.split(",");
                return Tuple3.of(fields[0], Integer.parseInt(fields[1]), fields[2]);
            }
        });

        //将正常的数据流和广播的数据流怎样才能发生关系（使用connect方法，可以共享状态）
        //将事实流（用户的行为流），与广播流进行connect
        SingleOutputStreamOperator<Tuple4<String, Integer, String, String>> res = tpStream2.connect(broadcastStream).process(new BroadcastProcessFunction<Tuple3<String, Integer, String>, Tuple3<String, String, String>, Tuple4<String, Integer, String, String>>() {

            //处理事实流的数据
            @Override
            public void processElement(Tuple3<String, Integer, String> input, ReadOnlyContext ctx, Collector<Tuple4<String, Integer, String, String>> out) throws Exception {
                ReadOnlyBroadcastState<String, String> broadcastState = ctx.getBroadcastState(stateDescriptor);
                String cid = input.f2;
                String name = broadcastState.get(cid);
                //输出数据
                out.collect(Tuple4.of(input.f0, input.f1, cid, name));
            }

            //处理广播流
            @Override
            public void processBroadcastElement(Tuple3<String, String, String> input, Context ctx, Collector<Tuple4<String, Integer, String, String>> out) throws Exception {
                System.out.println("processBroadcastElement invoked in subtask: " + getRuntimeContext().getIndexOfThisSubtask());
                String type = input.f2;
                String cid = input.f0;
                String cname = input.f1;
                BroadcastState<String, String> broadcastState = ctx.getBroadcastState(stateDescriptor);
                if ("DELETE".equals(type)) {
                    //移除原来的数据
                    broadcastState.remove(cid);
                } else {
                    //添加或更新数据
                    broadcastState.put(cid, cname);
                }
            }
        });

        res.print();

        env.execute();


    }
}
